package com.bjbn.consumer;

import com.bjbn.config.RocketMQConfig;
import com.bjbn.config.SystemConfig;
import com.bjbn.entity.FimsDataOri;
import com.bjbn.service.FimsDataService;
import com.bjbn.service.XmlParseService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;

/**
 * 全域机场运行管理平台RocketMQ消费者
 *
 * @author bjbn
 */
@Slf4j
@Component
public class AcdmRocketMQConsumer {

    @Autowired
    private RocketMQConfig rocketMQConfig;

    @Autowired
    private SystemConfig systemConfig;

    @Autowired
    private FimsDataService fimsDataService;

    @Autowired
    private XmlParseService xmlParseService;

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {

        // 检查是否为ACDM模式
        if (!"acdm".equals(systemConfig.getMode())) {
            log.info("当前不是ACDM模式，跳过ACDM消费者初始化");
            return;
        }

        log.info("初始化ACDM RocketMQ消费者...");
        RPCHook rpcHook = new AclClientRPCHook(
                new SessionCredentials(
                        rocketMQConfig.getAcdm().getConsumer().getKey(),
                        rocketMQConfig.getAcdm().getConsumer().getSecret()
                )
        );

        consumer = new DefaultMQPushConsumer(
                null,
                rocketMQConfig.getAcdm().getConsumer().getGroup(),
                rpcHook,
                new AllocateMessageQueueAveragely(),
                true,
                null
        );

        consumer.setNamesrvAddr(rocketMQConfig.getAcdm().getNameServer());
        String subExpression = rocketMQConfig.getAcdm().getConsumer().getSubExpression();
        consumer.subscribe(rocketMQConfig.getAcdm().getConsumer().getTopic(), subExpression);
        consumer.setUseTLS(true);
        consumer.setPollNameServerInterval(10000);

        // 设置消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String messageBody = new String(msg.getBody(), "UTF-8");
                        log.debug("收到ACDM消息: {}", messageBody);

                        // 解析XML并转换为FIMS数据
                        FimsDataOri fimsDataOri = xmlParseService.convertToFimsData(messageBody, "acdm");
                        if (fimsDataOri == null) {
                            log.warn("ACDM消息解析失败，跳过处理");
                            continue;
                        }

                        // 保存到MySQL
                        boolean mysqlResult = fimsDataService.saveFimsData(fimsDataOri);
                        if (!mysqlResult) {
                            log.error("ADCM数据保存到MySQL失败，FLID: {}", fimsDataOri.getFlid());
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }

                        // 推送到Kafka
                        boolean kafkaResult = fimsDataService.sendToKafka(fimsDataOri);
                        if (!kafkaResult) {
                            log.error("ACDM数据推送到Kafka失败，FLID: {}", fimsDataOri.getFlid());
                        }

                        log.debug("ACDM消息处理完成，FLID: {}", fimsDataOri.getFlid());

                    } catch (Exception e) {
                        log.error("处理ACDM消息异常", e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        log.debug("ACDM RocketMQ消费者启动成功，Topic: {}, Group: {}",
                rocketMQConfig.getAcdm().getConsumer().getTopic(),
                rocketMQConfig.getAcdm().getConsumer().getGroup());
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("ACDM RocketMQ消费者已关闭");
        }
    }
} 